package com.zero.rocketmq.api.base.customer;

import com.zero.rocketmq.api.utils.RocketMQContents;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

public class LitePullConsumerAssign {

    public static volatile boolean running = true;

    public static void main(String[] args) throws MQClientException {

        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(RocketMQContents.CUSTOMER_GROUP);
        consumer.setNamesrvAddr(RocketMQContents.NAMESERVER_ADDR);
        consumer.setAutoCommit(false);
        consumer.start();

        Collection<MessageQueue> messageQueues = consumer.fetchMessageQueues(RocketMQContents.TOPIC);
        List<MessageQueue> list = new ArrayList<>(messageQueues);
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0;i<list.size()/2;i++){
            assignList.add(list.get(i));
        }

        consumer.assign(assignList);
        consumer.seek(assignList.get(0), 10);

        try{
            while (running){
                List<MessageExt> messageExts = consumer.poll();
                System.out.printf("%s %n", messageExts);
                consumer.commitSync();
            }
        }finally {
            consumer.shutdown();
        }


    }
}
